package async;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.mina.filter.codec.protobuf.TestProto;

import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.Message;

//http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6446653
public class NioSender {

	private Executor executor;
	
	private final Selector selector;
	private final SocketChannel channel;
	private final Object gate = new Object();
	
	private final Queue<Message> sendQueue = new ConcurrentLinkedQueue<Message>();
	private final AtomicInteger counter = new AtomicInteger();
	
	public NioSender() throws IOException {
		selector = Selector.open();
		channel = SocketChannel.open();
		channel.configureBlocking(false);
	}
	
	public void connect() throws IOException {
		System.out.println("Connect...");
		if(!channel.isConnected()) {
			if(!channel.connect(new InetSocketAddress(hostname, port))) {
		        synchronized (gate) {
					System.out.println("Wake up selector!");
					selector.wakeup();
					System.out.println("Register selector for OP_CONNECT");
					channel.register(selector, SelectionKey.OP_CONNECT);
		        }
			} else {
				System.out.println("Connected: " + channel);
			}
		}
	}
	
	public void send(Message message) throws IOException {
		System.out.println("Send...");
		if(sendQueue.offer(message)) {
			counter.incrementAndGet();
		}
	}
	
	public static void main(String[] args) throws Exception {
		NioSender ns = new NioSender();
		ns.hostname = "localhost";
		ns.port = 6667;
		ns.executor = Executors.newFixedThreadPool(10);
		ns.start();
		
		ns.connect();
		for(int i = 0; i < 100; i++) {
			ns.send(TestProto.Foo.newBuilder().setFoo(new Random().nextInt(64 * 1024)).build());	
		}
	}
	
	public void start() throws IOException {
		executor.execute(new Runnable() {
			public void run() {
				while(true) {
					try {
						//System.out.println("Select...");
						int n = selector.select();
						//System.out.println("Selected: " + n);
					} catch (IOException e) {
						throw new RuntimeException(e);
					}
					for(Iterator<SelectionKey> keys = selector.selectedKeys().iterator(); keys.hasNext(); ) {
						final SelectionKey key = keys.next();
						if(key.isValid()) {
							if(key.isConnectable()) {
								executor.execute(new Runnable() {
									public void run() {
										try {
											System.out.println("To connect: " + channel);
											if(!channel.isConnected() && channel.isConnectionPending()) {
												channel.finishConnect();
												System.out.println("Register selector for OP_WRITE");
												channel.keyFor(selector).interestOps(SelectionKey.OP_WRITE);
												System.out.println("Connected: " + channel);
											}
										} catch (IOException e) {
											throw new RuntimeException(e);
										}
									}});
							}
							if(key.isWritable()) {
								executor.execute(new Runnable() {
									public void run() {
										try {
												
												Message message = sendQueue.poll();
												
												if(message != null) {
													
													System.out.println("To write: " + message);
													System.out.println("Counter: " + counter.decrementAndGet());
													
													//prepare buffer
													int messagelength = message.getSerializedSize();
													int prefixLength = CodedOutputStream.computeRawVarint32Size(messagelength);
													int totalLength = messagelength +  prefixLength;
													byte[] buf = new byte[totalLength];
													
													//write length
													CodedOutputStream cos = CodedOutputStream.newInstance(buf);
													cos.writeRawVarint32(messagelength);
													cos.flush();
													
													//write buffer
													ByteBuffer b = ByteBuffer.wrap(buf);
													b.position(prefixLength);
													b.put(message.toByteArray());
													b.flip();
													int n = channel.write(b);
													
													System.out.println("Wrote: " + n);	
												}
												
										} catch (IOException e) {
											throw new RuntimeException(e);
										}
									}});
							}
						}
					}
					synchronized (gate) { /* do nothing */ }
				}
			}});
	}
	
}
